package com.xueshan.disruptor.lmax.queue;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.xueshan.disruptor.lmax.consumer.DisruptorConsumer;
import com.xueshan.disruptor.lmax.event.ObjectEvent;
import com.xueshan.disruptor.lmax.event.ObjectEventFactory;

import java.util.concurrent.Executors;

/**
 * created with IDEA
 *
 * @author:huqm
 * @date:2021/3/16
 * @time:17:26 <p>
 *
 * </p>
 */
public class DisruptorQueueFactory {
    public DisruptorQueueFactory() {
    }



    // 创建"发布订阅模式"的操作队列，即同一事件会被多个消费者并行消费
    public static <T> DisruptorQueue<T> getHandleEventsQueue(int queueSize, boolean isMoreProducer,
                                                             DisruptorConsumer<T>... consumers) {
        Disruptor<ObjectEvent<T>> disruptor = new Disruptor(new ObjectEventFactory(),
                queueSize, Executors.defaultThreadFactory(),
                isMoreProducer ? ProducerType.MULTI : ProducerType.SINGLE,
                new YieldingWaitStrategy());
        disruptor.handleEventsWith(consumers);
        return new DisruptorQueue(disruptor);
    }

    // 直接通过传入的 Disruptor 对象创建操作队列（如果消费者有依赖关系的话可以用此方法）
    public static <T> DisruptorQueue<T> getQueue(Disruptor<ObjectEvent<T>> disruptor) {
        return new DisruptorQueue(disruptor);
    }
}

class EventExceptionHandler implements ExceptionHandler {

    @Override
    public void handleEventException(Throwable ex, long sequence, Object event) {
        System.out.println("handleEventException：" + ex);
    }

    @Override
    public void handleOnShutdownException(Throwable ex) {
        System.out.println("handleEventException：" + ex);
    }

    @Override
    public void handleOnStartException(Throwable ex) {
        System.out.println("handleOnStartException：" + ex);
    }
}
